package com.zjc.springboot.kafka.consumer;

import com.zjc.springboot.kafka.ChannelDefinition;
import com.zjc.springboot.kafka.messages.StormMsg;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.function.Consumer;

@Component
@EnableBinding(ChannelDefinition.class)
public class KafkaConsumer {

    private static Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);

    /**
     * stream 3.0 之前
     * @param message
     */
    /*@StreamListener(ChannelDefinition.TEST_STORM_INPUT)
    public void testStormInput(Message<StormMsg> message) {
        logger.info("logger testStormInput message receive: {}" + message.getPayload());
    }*/

    /**
     * stream 3.0 之后
     * @return
     */
    @Bean
    Consumer<StormMsg> testStorm() {

        return stormMsg -> {

            stormMsg.getValue();
            logger.info("testStormInput message receive: {}" + stormMsg);
        };
    }
}
